package com.cml.client.domain.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cml.client.common.config.Configuration;
import com.cml.client.common.context.ApplicationContext;
import com.cml.client.domain.entity.Server;
import com.cml.client.infrastructure.MessageQueueManager;
import com.cml.client.infrastructure.NetworkManager;
import com.cml.common.dto.entity.Message;
import com.cml.common.dto.entity.ServerNode;
import com.cml.common.dto.entity.ServiceInstance;
import com.cml.common.dto.entity.SoltMetadata;
import com.cml.common.dto.message.*;
import com.cml.common.util.SoltMathUtil;
import org.apache.log4j.Logger;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 姓名：陈茂林
 * 时间：2023/10/16 14:37
 * 描述：
 */
public class ServiceInstanceService {
    private  NetworkManager networkManager;

    private Configuration configuration = Configuration.getInstance();

    private Logger logger = Logger.getLogger(ServiceInstanceService.class);

    private Server followerServer;

    private ServerNode routeServerNode;

    private Map<String,String> chacheSubscribeList=new HashMap<>();

    private BlockingQueue subscribeServiceInstanceQueue = new LinkedBlockingQueue();



    private Map serverList =new HashMap();

    private List<SoltMetadata> slotsAllocation =
            new ArrayList();

    public ServiceInstanceService(ApplicationContext applicationContext){
        this.networkManager = applicationContext.getNetworkManager();
    }
    public void start() throws Exception {
        initServerInstance();
        //启动心跳
        new HearbeatThread().start();
        //同步实例变化
        new ServiceInstanceChangeSynThread().start();

    }

    public void initServerInstance()throws Exception{
        if(chooseFollowerConn()) {
            //抓取所有的槽位信息
            fetchSlotsAllocation();
            //获取所有服务列表
            fetchServerList();
            //分配当前client的server
            routeServer();

            registerToServer();
        }else{
            throw new RuntimeException("无法连接服务管理器");
        }
    }



    class HearbeatThread extends Thread{
        @Override
        public void run() {
            ServiceHearbeatRequest message = new ServiceHearbeatRequest();
            message.setServiceName(Configuration.getInstance().getServiceName());
            Server server = Server.Builder.build(routeServerNode);
            while(true){
                try{

                    MessageQueueManager.sendRequestMessage(server.getServerConnDesc(),message);
                    logger.info("发送心跳消息:"+server.getServerConnDesc());
                    Thread.sleep(configuration.getServiceHeartbeatIntervalTime());
                }catch (Exception e){
                    logger.info("心跳消息发送异常:"+server.getServerConnDesc());
                    logger.error(e.getMessage(),e);
                }
            }
        }
    }



    public void registerToServer() throws Exception {
        ServiceRegisterRequest message = new ServiceRegisterRequest();
        message.setServiceName(Configuration.getInstance().getServiceName());
        message.setHttpPort(Configuration.getInstance().getHttpPort());
        Server server = Server.Builder.build(this.routeServerNode);
        networkManager.connServiceServer(server);
        logger.info("开始注册服务信息，服务名称为："+message.getServiceName());
        MessageQueueManager.sendRequestMessage(server.getServerConnDesc(),message);

        BlockingQueue blockingQueue = new LinkedBlockingQueue<>();
        MessageQueueManager.registerDataExchangeQueue(ServiceRegisterResponse.class,blockingQueue);
        blockingQueue.take();
        MessageQueueManager.removeDataExchangeQueue(ServiceRegisterResponse.class);
        logger.info("服务注册成功！");
    }

    public void subscribe(String serviceName)throws Exception {
        if(chacheSubscribeList.get(serviceName) == null){
            ServiceSubscribeRequest serviceSubscribeRequest = new ServiceSubscribeRequest();
            serviceSubscribeRequest.setServiceName(serviceName);

            ServerNode subscribeServerNode = getServiceNameServerNode(serviceName);
            Server server = Server.Builder.build(subscribeServerNode);
            networkManager.connServiceServer(server);
            MessageQueueManager.sendRequestMessage(server.getServerConnDesc(),serviceSubscribeRequest);
            chacheSubscribeList.put(serviceName,serviceName);
            logger.info(server.getServerConnDesc() +" 订阅服务"+serviceName+"完成");
            MessageQueueManager.registerDataExchangeQueue(ServiceInstanceChangeRequest.class,subscribeServiceInstanceQueue);
        }
    }

    private void  routeServer(){
        routeServerNode = getServiceNameServerNode(Configuration.getInstance().getServiceName());
        logger.info("当前注册服务器为:"+routeServerNode);
    }

    private ServerNode getServiceNameServerNode(String serviceName){
        Integer slot = SoltMathUtil.mathSolt(serviceName);
        ServerNode routeServerNode = (ServerNode) this.serverList.get(this.locateServerBySlot(slot));
        return routeServerNode;
    }

    private Integer locateServerBySlot(Integer slot) {
        Integer serverNodeId = null;
        for(SoltMetadata soltMetadata:slotsAllocation){
            if(slot >=soltMetadata.getStartSlot() && slot <=soltMetadata.getEndSlot()){
                serverNodeId=soltMetadata.getServerNodeId();
            }

        }
        return serverNodeId;
    }


    private void fetchServerList() throws InterruptedException {
        Message message = new FetchAllServerListRequest();
        MessageQueueManager.sendRequestMessage(this.followerServer.getServerConnDesc(),message);

        BlockingQueue blockingQueue = new LinkedBlockingQueue<>();
        MessageQueueManager.registerDataExchangeQueue(FetchAllServerListReponse.class,blockingQueue);
        FetchAllServerListReponse fetchAllServerListReponse = (FetchAllServerListReponse) blockingQueue.take();
        MessageQueueManager.removeDataExchangeQueue(FetchAllServerListReponse.class);

        logger.info("获取到所有的server列表为："+fetchAllServerListReponse.getAllServerList());
        String allServerList = fetchAllServerListReponse.getAllServerList();
        Map remoteServerMap =JSONObject.parseObject(allServerList,Map.class);
        Set<Integer> remoteServerSet = remoteServerMap.keySet();
        for (Integer key : remoteServerSet){
            JSONObject jsonObject = (JSONObject) remoteServerMap.get(key);
            ServerNode serverNode = jsonObject.toJavaObject(ServerNode.class);
            this.serverList.put(key,serverNode);
        }

    }

    private boolean chooseFollowerConn(){
        boolean isConn = false;
        List<Map> nodeFollowerServerList=configuration.getNodeFollowerServerList();

        for(Map nodeFollowerServer : nodeFollowerServerList){
            try {
                Server server = new Server();
                server.setIp((String) nodeFollowerServer.get("ip"));
                server.setPort(Integer.parseInt((String) nodeFollowerServer.get("port")));
                this.networkManager.connServiceServer(server);
                followerServer=server;

                logger.info("当前连接服务为："+nodeFollowerServer);
                isConn=true;
                break;
            }catch (Exception e){
                logger.error("server连接失败："+nodeFollowerServer);
                logger.error(e);
            }
        }
        return isConn;
    }

    private void fetchSlotsAllocation() throws InterruptedException {
        Message message = new FetchSlotsAllocationRequst();
        MessageQueueManager.sendRequestMessage(this.followerServer.getServerConnDesc(),message);

        BlockingQueue blockingQueue = new LinkedBlockingQueue<>();
        MessageQueueManager.registerDataExchangeQueue(FetchSlotsAllocationResponse.class,blockingQueue);
        FetchSlotsAllocationResponse fetchSlotsAllocationResponse = (FetchSlotsAllocationResponse) blockingQueue.take();
        MessageQueueManager.removeDataExchangeQueue(FetchSlotsAllocationResponse.class);

        String soltMetadataList=fetchSlotsAllocationResponse.getSoltMetadataList();
        slotsAllocation= JSONObject.parseArray(soltMetadataList,SoltMetadata.class);
        logger.info("获取所有的槽位信息为："+soltMetadataList);
    }



    public List<ServiceInstance> getServiceInstanceList(String serviceName){
        return CacheRegisterServiceInstance.getInstance().getServiceInstanceList(serviceName);
    }

    class ServiceInstanceChangeSynThread extends  Thread{
        @Override
        public void run() {
            while (true){
                try {
                    ServiceInstanceChangeRequest serviceInstanceChangeRequest = (ServiceInstanceChangeRequest) subscribeServiceInstanceQueue.take();
                    logger.info(serviceInstanceChangeRequest);
                    JSONArray serviceInstanceJsonArray = JSONArray.parseArray(serviceInstanceChangeRequest.getServiceInstanceRegisterList());
                    for (int i = 0; i < serviceInstanceJsonArray.size(); i++) {
                        JSONObject serviceInstance = (JSONObject) serviceInstanceJsonArray.get(i);
                        CacheRegisterServiceInstance.getInstance().serviceInstanceCache(new ServiceInstance.Builder().build(serviceInstance));

                    }
                } catch (InterruptedException e) {
                    logger.error(e);
                }
            }
        }
    }
}
